package consys.social.core.dao;

import java.util.Comparator;
import java.util.Collections;
import java.util.Iterator;
import java.util.Map;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Delete;
import java.util.ResourceBundle;
import org.apache.log4j.Logger;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import consys.social.core.bo.Entity;
import consys.social.core.bo.Wall;
import java.util.ArrayList;
import consys.social.core.bo.WallPost;
import consys.social.core.bo.WallPostType;
import consys.social.core.service.SocialUtils;
import consys.social.memcached.dao.spymemcached.Memcached;
import consys.social.memcached.service.spymemcached.WallPostTranscoder;
import consys.social.nosql.dao.hbase.HBase;
import java.io.IOException;
import java.text.ParseException;
import java.util.Date;
import java.util.List;
import org.apache.hadoop.hbase.client.Put;
import static consys.social.nosql.service.hbase.HBaseHelpers.*;

/**
 *
 * @author petan
 */
public class WallDaoImpl implements WallDao {

    private Memcached memcached;
    private ThreadLocal<HBase> walls;
    private ThreadLocal<HBase> discussions;
    private int storeSeconds;
    private int holdDiscPosts = Integer.parseInt(ResourceBundle.getBundle("social").getString("post.discussion.holdInMem"));
    private Logger logger = Logger.getLogger(WallDaoImpl.class);

    public void setStoreSeconds(int storeSeconds) {
        this.storeSeconds = storeSeconds;
    }

    public void setMemcached(Memcached memcached) {
        this.memcached = memcached;
    }

    public void setWalls(ThreadLocal<HBase> walls) {
        this.walls = walls;
    }

    public void setDiscussions(ThreadLocal<HBase> discussions) {
        this.discussions = discussions;
    }

    @Override
    public void savePost(List<String> followersUids, WallPost post) throws IOException {

        byte[] key = SocialUtils.generateReverseKey(post.getAddedBy().getUid(), post.getDate());
        String memcachedKey = SocialUtils.generateKey(post.getAddedBy().getUid(), post.getDate());
        try {
            Put put = new Put(key);
            addToPut(put, "text", "", post.getText());
            addToPut(put, "info", "owner", post.getAddedBy().getUid());
            addToPut(put, "info", "likesCount", (long) 0);
            addToPut(put, "info", "discussionCount", (long) 0);
            addToPut(put, "info", "type", post.getType().getTypeNumber());
            if (walls.get().add(put)) {
                memcached.add(memcachedKey, post, storeSeconds, new WallPostTranscoder());
                for (String follower : followersUids) {
                    List<String> wall = (List<String>) memcached.get(follower);
                    if (wall == null) {
                        wall = new ArrayList<String>();
                    }
                    wall.add(0, memcachedKey);
                    memcached.set(follower, wall, 0);
                }
            }
        } catch (IOException ex) {
            logger.error("HBase problem - saving post");
            throw ex;
        }
    }

    @Override
    public void deletePost(WallPost post) throws IOException {
        byte[] key = SocialUtils.generateReverseKey(post.getAddedBy().getUid(), post.getDate());
        String memcachedKey = SocialUtils.generateKey(post.getAddedBy().getUid(), post.getDate());
        try {
            Get get = new Get(key);
            addToGet(get, "text", "");
            Delete del = new Delete(key);
            if (walls.get().deleteIfGet(get, del)) {
                memcached.delete(memcachedKey);
            }
        } catch (IOException ex) {
            logger.error("HBase problem - deleting post");
            throw ex;
        }

    }

    @Override
    public Wall loadWall(Entity entity, int count, int from) {
        logger.info("Loading Wall");
        Wall wall = new Wall();
        List<String> postLinks = (List<String>) memcached.get(entity.getUid());
        

        if (postLinks == null) {
            postLinks = new ArrayList<String>();
        }

        Map<String, WallPost> wallPosts = memcached.getBulk(postLinks.subList(from, Math.min(from+count, postLinks.size())), new WallPostTranscoder());

        for (Iterator<Map.Entry<String, WallPost>> i = wallPosts.entrySet().iterator(); i.hasNext();) {
            Map.Entry<String, WallPost> entry = i.next();
            if (entry.getValue() == null) {
                i.remove();
            }
        }

        memcached.replace(entity.getUid(), new ArrayList(wallPosts.keySet()), 0);
        List<WallPost> wpsts = new ArrayList(wallPosts.values());
        Collections.sort(wpsts, new Comparator<WallPost>() {
            @Override
            public int compare(WallPost t, WallPost t1) {
                return t.compareTo(t1);
            }
        });
        wall.setWallPosts(wpsts);
        return wall;
    }

    @Override
    public Wall loadProfileWall(Entity entity, int count, Date from, boolean newest) throws IOException {
        logger.info("Loading profile wall");
        Wall wall = new Wall();
        Scan s = new Scan(SocialUtils.generateReverseKey(entity.getUid(), from), SocialUtils.generateReverseKeyMax(entity.getUid()));
        addToScan(s, "text", "");
        addToScan(s, "info", "likesCount");
        addToScan(s, "info", "discussionCount");
        addToScan(s, "info", "owner");
        addToScan(s, "info", "type");
        int loaded = 0;
        try {
            ResultScanner rs = walls.get().getScanner(s);
            if (!newest) {
                rs.next();
            }
            for (Result r : rs) {
                loaded++;
                wall.addWallPost(loadPostFromHBase(r));
                if (loaded >= count) {
                    return wall;
                }
            }
        } catch (IOException ex) {
            logger.error("HBase problem - loading profile wall");
            throw ex;
        }
        return wall;
    }

    private WallPost loadPostFromHBase(Result r, boolean minified) throws IOException {
        WallPost wp = new WallPost();
        wp.setAddedBy(new Entity(Bytes.toString(getColumnValue(r, "info", "owner"))));
        wp.setText(Bytes.toString(getColumnValue(r, "text", "")));
        wp.setLikesCount(Bytes.toLong(getColumnValue(r, "info", "likesCount")));
        wp.setDiscussionCount(Bytes.toLong(getColumnValue(r, "info", "discussionCount")));
        wp.setType(new WallPostType(Bytes.toInt(getColumnValue(r, "info", "type"))));
        try {
            wp.setDate(SocialUtils.getKeyDate(r.getRow()));
        } catch (ParseException E) {
            //should never happen
            logger.error("Error parsing date");
        }
        if (!minified) {
            //load even more - than needed just on the wall (discussion, who likes and so on);
        } else {
            String stringKey = SocialUtils.generateKey(wp.getAddedBy().getUid(), wp.getDate());
            Scan s = new Scan(SocialUtils.generateReverseKey(stringKey, new Date()), SocialUtils.generateReverseKeyMax(stringKey));
            int loaded = 0;
            try {
                ResultScanner rs = discussions.get().getScanner(s);
                for (Result rr : rs) {
                    loaded++;
                    wp.addDiscussionPost(new DiscussionLoader().load(rr));
                    if (loaded >= holdDiscPosts) {
                        break;
                    }
                }
            } catch (IOException ex) {
                logger.error("Error loading post from HBase");
                throw ex;
            }
        }
        return wp;
    }

    private WallPost loadPostFromHBase(Result r) throws IOException {
        return loadPostFromHBase(r, true);
    }
}